package com.changzhi.common.mq;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.BooleanUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.changzhi.agora.AgoraUtil;
import com.changzhi.common.constans.Constants;
import com.changzhi.common.constants.RedisConstants;
import com.changzhi.common.enums.LangEnum;
import com.changzhi.common.enums.OrderStatusTypeEnum;
import com.changzhi.common.enums.StatusCodeEnum;
import com.changzhi.common.util.MD5Util;
import com.changzhi.common.util.Result;
import com.changzhi.ldyl.model.*;
import com.changzhi.ldyl.model.dto.InviteDto;
import com.changzhi.ldyl.model.dto.SmsDto;
import com.changzhi.mapper.ActivityLogMapper;
import com.changzhi.mapper.ChannelVIPMapper;
import com.changzhi.mapper.PaymentOrderMapper;
import com.changzhi.service.*;
import com.changzhi.util.ObjectAndByteUtil;
import com.changzhi.util.SmsGlobeUtil;
import com.changzhi.util.SmsHuaweiUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;

@Slf4j
@Component
public class MqListener {

    @Autowired
    private OrderDayService orderDayService;
    @Resource
    private ActivityLogMapper activityLogMapper;
    @Autowired
    private UserGiftDayService userGiftDayService;
    @Autowired
    private TanabataService tanabataService;
    @Autowired
    private SignService signService;
    @Autowired
    private InviteService inviteService;
    @Autowired
    private UserService userService;
    @Autowired
    private PaymentOrderMapper paymentOrderMapper;

    @Autowired
    private PaymentOrderNotifyQueueService paymentOrderNotifyQueueService;

    @Autowired
    private PrivateChatMsgService privateChatMsgService;

    @Autowired
    private RedService redService;
    @Resource
    private ChannelVIPService ChannelVIPService;


    /**
     * 延时Queue
     */
    public static final String DELAY_QUEUE = "dead.queue";

    @RabbitListener(queues = DELAY_QUEUE)
    public void process(Message message, @Headers Map<String,Object> headers, Channel channel) throws IOException {
        /*try {
            PaymentOrder paymentOrder = (PaymentOrder) ObjectAndByteUtil.toObject(message.getBody());
            log.info("methodName:createPaymentOrderNotify,接收消息时间：{}，消息内容：{}", DateUtil.now(),JSON.toJSONString(paymentOrder));

            PaymentOrder order = paymentOrderMapper.selectOne(new QueryWrapper<PaymentOrder>()
                    .eq(PaymentOrder.ORDER_NO, paymentOrder.getOrderNo()).last("limit 1"));
            if (order != null) {
                if (OrderStatusTypeEnum.WAIT_BUYER_PAY.getDesc().equals(order.getOrderStatus())) {
                    User user = userService.getUserByAccountId(10000);
                    if (null == user) {
                        log.error("delay.queue.paymentOrder====10000号不存在");
                    } else {
                        String userId = user.getUserId();
                        String destinationUserId = paymentOrder.getCliBuyerId();
                        User destinationUser = userService.getUser(destinationUserId);
                        String lang = destinationUser.getLang();
                        LambdaQueryWrapper<PaymentOrderNotifyQueue> queueLambdaQueryWrapper = new LambdaQueryWrapper<>();
                        queueLambdaQueryWrapper.eq(PaymentOrderNotifyQueue::getCliBuyerId,destinationUserId);
                        queueLambdaQueryWrapper.eq(PaymentOrderNotifyQueue::getStatus,1);
                        List<PaymentOrderNotifyQueue> list = paymentOrderNotifyQueueService.list(queueLambdaQueryWrapper);
                        if (CollUtil.isEmpty(list)) {
                            String textCN = "Hello，我是Boba支付客服，支付问题可以直接咨询我。\n" +
                                    "如果回复不及时，也可以联系以下客服：\n" +
                                    "（赖）Line：boba\n" +
                                    "（微）WeChat：adkovip";

                            String textTW = "Hello，我是Boba支付客服，支付問題可以直接諮詢我。\n" +
                                    "如果回覆不及時，也可以聯絡以下客服：\n" +
                                    "（賴）Line：boba\n" +
                                    "（微）WeChat：adkovip";

                            String textEN = "Hello, I am Boba payment customer service, you can directly consult me for payment issues.\n" +
                                    "If the reply is not timely, you can also contact the following customer service:\n" +
                                    "Line: boba\n" +
                                    "WeChat: adkovip";

                            String text;
                            if (StrUtil.equals(LangEnum.ZH_CN.getLang(),lang)) {
                                text = textCN;
                            } else if (StrUtil.equals(LangEnum.ZH_TW.getLang(),lang)) {
                                text = textTW;
                            } else {
                                text = textEN;
                            }
                            log.info("点对点消息发送开始：用户id：{}",destinationUserId);
                            AgoraUtil.peerMessages(userId,destinationUserId,text);
                            log.info("点对点消息发送结束");
                            LambdaUpdateWrapper<PaymentOrderNotifyQueue> wrapper = new LambdaUpdateWrapper<>();
                            wrapper.eq(PaymentOrderNotifyQueue::getOrderNo,paymentOrder.getOrderNo()).set(PaymentOrderNotifyQueue::getStatus,1);
                            paymentOrderNotifyQueueService.update(wrapper);

                            PrivateChatMsg msg = new PrivateChatMsg();
                            String privateChannelId;
                                if (user.getUserId().compareTo(destinationUser.getUserId()) > 0) {
                                    privateChannelId = user.getUserId() + destinationUser.getUserId();
                                } else {
                                    privateChannelId = destinationUser.getUserId() + user.getUserId();
                                }
                                privateChannelId = MD5Util.getMD5Str(privateChannelId);
                            msg.setPrivateChannelId(privateChannelId);
                            msg.setSpeakerUserId(user.getUserId());
                            msg.setSpeakerAccountId(user.getAccountId());
                            msg.setListenerUserId(destinationUser.getUserId());
                            msg.setListenerAccountId(destinationUser.getAccountId());
                            msg.setContent(text); // 需要将原始内容保存到数据库
                            msg.setStatus(1);
                            msg.setCreateTime(LocalDateTime.now());
                            boolean saved = privateChatMsgService.save(msg);
                        }
                    }
                } else {
                    LambdaUpdateWrapper<PaymentOrderNotifyQueue> wrapper = new LambdaUpdateWrapper<>();
                    wrapper.eq(PaymentOrderNotifyQueue::getOrderNo,paymentOrder.getOrderNo()).set(PaymentOrderNotifyQueue::getStatus,2);
                    paymentOrderNotifyQueueService.update(wrapper);
                }
            }

            //手动ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            log.error("点对点消息发送失败:{}",e.getMessage());
            //手动ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }*/
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }







    /**
     * 赠送礼物消费者
     * @param msg
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "#{userGiftQueue.name}", durable = "true"),
            exchange = @Exchange(value = "user.gift.fanout.exchange", type = ExchangeTypes.FANOUT)
    ))
    public void userGiftListen(String msg,@Headers Map<String,Object> headers, Channel channel) throws Exception {
        log.info("userGiftListen 接收到消息：{}", msg);
        UserGift userGift = JSON.parseObject(msg, UserGift.class);
        if(userGift == null){
            return;
        }
      /*  //防重判断
        QueryWrapper<ActivityLog> wrapper = new QueryWrapper<>();
        wrapper.lambda().eq(ActivityLog::getSourceId, userGift.getId());
        ActivityLog activityLog = activityLogMapper.selectOne(wrapper);
        if(activityLog != null){
            log.error("已有该数据！,userGift={}", JSON.toJSONString(userGift));
            //return;
        } else {
            //插入日志表
            ActivityLog record = new ActivityLog();
            record.setActivityId(ACTIVITY_ID);
            record.setChannelId(userGift.getChannelId());
            record.setSourceId(userGift.getId());
            activityLogMapper.insert(record);
        }*/
        //七夕活动
        try {
            tanabataService.doMqDataByUserGift(userGift);
        } catch (Exception e) {
            log.error("methodName:userGiftListen,周星活动,userGift={},error={}", JSON.toJSONString(userGift), e.getMessage());
            e.printStackTrace();
        }
        //星座活动
        try {
            signService.doMqDataByUserGift(userGift);
        } catch (Exception e) {
            log.error("methodName:userGiftListen,星座活动,userGift={},error={}", JSON.toJSONString(userGift), e.getMessage());
            e.printStackTrace();
        }

        try {
            redService.doMqDataByUserGift(userGift);
        } catch (Exception e) {
            log.error("methodName:userGiftListen,圣诞节活动,userGift={},error={}", JSON.toJSONString(userGift), e.getMessage());
            e.printStackTrace();
        }

        try {
            ChannelVIPService.checkChannelVip(userGift);
        } catch (Exception e) {
            log.error("methodName:userGiftListen,检查房间VIP,userGift={},error={}", JSON.toJSONString(userGift), e.getMessage());
            e.printStackTrace();
        }

        //统计
        try {
            userGiftDayService.doMqDataByHandler(userGift);
        } catch (Exception e) {
            log.error("methodName:userGiftListen,统计,userGift={},error={}", JSON.toJSONString(userGift), e.getMessage());
            e.printStackTrace();
        }
        //手动ack
        Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliveryTag, false);
    }

    /**
     * 订单支付成功消费者
     * @param msg
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "#{orderSuccessQueue.name}", durable = "true"),
            exchange = @Exchange(value = "order.success.fanout.exchange", type = ExchangeTypes.FANOUT)
    ))
    public void orderSuccessListen(String msg,@Headers Map<String,Object> headers, Channel channel) throws Exception {
        log.info("orderSuccessListen 接收到消息：{}", msg);

        try {
            orderDayService.doMqDataByOrderSuccess(msg);
        } catch (Exception e) {
            log.error("doMqDataByOrderSuccess msg:{}",JSON.toJSONString(msg),e);
        }

        try {
            orderDayService.sendFirstRechargeAward(msg);
        } catch (Exception e) {
            log.error("sendFirstRechargeAward msg:{}",JSON.toJSONString(msg),e);
        }

        try {
            orderDayService.checkUserPaymentChannel(msg);
        } catch (Exception e) {
            log.error("checkUserPaymentChannel msg:{}",JSON.toJSONString(msg),e);
        }

        try {
            orderDayService.checkAccount10000Follower(msg);
        } catch (Exception e) {
            log.error("checkAccount10000Follower msg:{}",JSON.toJSONString(msg),e);
        }

        //手动ack
        Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliveryTag, false);

    }

    /**
     * pk结束消费者
     * @param msg
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "#{pkFinishQueue.name}", durable = "true"),
            exchange = @Exchange(value = "pk.finish.fanout.exchange", type = ExchangeTypes.FANOUT)
    ))
    public void pkFinishListen(String msg,@Headers Map<String,Object> headers, Channel channel) throws Exception {
        log.info("pkFinishListen 接收到消息：{}", msg);
        //yearService.doMqDataByPkFinish(msg);
        //手动ack
        Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliveryTag, false);
    }

    /**
     * 绑定手机号
     * @param msg
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "#{bindMobileQueue.name}", durable = "true"),
            key = { Constants.USER_BIND_MOBILE},
            exchange = @Exchange(value = "user.invite.fanout.exchange")
    ))
    public void bindMobileListen(String msg,@Headers Map<String,Object> headers, Channel channel) throws Exception {
        log.info("bindMobileListen 接收到消息：{}", msg);
        InviteDto inviteDto = JSON.parseObject(msg, InviteDto.class);
        if(inviteDto == null){
            return;
        }
        inviteService.doMqDataByHandler(inviteDto);
        //手动ack
        Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliveryTag, false);
    }


    /**
     * 用户填写邀请码(奶爸邀请用户)
     * @param msg
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "#{userInviteQueue.name}", durable = "true"),
            key = {Constants.USER_INVITE_KEY},
            exchange = @Exchange(value = "user.invite.fanout.exchange")
    ))
    public void userInviteListen(String msg,@Headers Map<String,Object> headers, Channel channel) throws Exception {
        log.info("userInviteListen 接收到消息：{}", msg);
        inviteService.doMqDataByUserInvite(msg);
        //手动ack
        Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliveryTag, false);
    }

    /**
     * 短信
     * @param msg
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "#{smsQueue.name}", durable = "true"),
            exchange = @Exchange(value = "sms.fanout.exchange", type = ExchangeTypes.FANOUT)
    ))
    public void smsListen(String msg,@Headers Map<String,Object> headers, Channel channel) throws Exception {
        log.info("smsListen 接收到消息：{}", msg);
        SmsDto smsDto = JSON.parseObject(msg, SmsDto.class);
        if(smsDto == null){
            return;
        }
        boolean b = checkIsCn(smsDto.getMobile());
        if (BooleanUtil.isTrue(b)) {
            SmsHuaweiUtil.sendMsg(smsDto.getMobile(),smsDto.getCode());
        } else {
            SmsGlobeUtil.sendMsg(smsDto.getMobile(),smsDto.getCode().toString());
        }
        //手动ack
        Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliveryTag, false);
    }

    private boolean checkIsCn(String mobile) {
        boolean isCn86 = StrUtil.startWith(mobile, "86");
        if (isCn86) {
            String subPhone = StrUtil.subSuf(mobile, 2);
            isCn86 = StrUtil.length(mobile) == 13 && subPhone.matches("^(1)(\\d{10})$");
            if (BooleanUtil.isFalse(isCn86)) {
                return false;
            }
        } else {
            isCn86 = StrUtil.length(mobile) == 11 && mobile.matches("^(1)(\\d{10})$");
        }
        return isCn86;
    }



    /**
     * 活动奖励发放
     * @param msg
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "#{activityAwardQueue.name}", durable = "true"),
            key = { Constants.AWARD_KEY},
            exchange = @Exchange(value = "activity.info.direct.exchange")
    ))
    public void activityAwardListen(String msg,@Headers Map<String,Object> headers, Channel channel) throws Exception {
        log.info("activityAwardListen 接收到消息：{}", msg);
        try {
            redService.doMqDataByActivityAward(msg);
        } catch (Exception e) {
            log.error("methodName:activityAwardListen,圣诞节活动,msg={},error={}", msg, e.getMessage());
            e.printStackTrace();
        }
        //手动ack
        Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliveryTag, false);
    }


}